package cn.wud.tags.hbase.tools

import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions.{col, udf}

/**
 *
 * @author wudl
 * @create 2021/7/12 15:45
 * @description 针对标签进行相关操作的工具类
 */
object TagTools {

  /**
   * 将标签数据中的属性标签规则rule拆分为范围start ,end
   *
   * @param tagDF
   * @return
   */

  def convertTuple(tagDF: DataFrame): DataFrame = {
    import tagDF.sparkSession.implicits._
    import org.apache.spark.sql.functions._
    //1. 自定义udf 函数 ，解析标签规则rule
    val rule_to_tuple: UserDefinedFunction = udf(
      (rule: String) => {
        val Array(start, end) = rule.split("-").map(_.toInt)
        //  返回一个二元组
        (start, end)
      }
    )
    //2. 获取属性标签数据，解析规则rule
    val ruleDF: DataFrame = tagDF.filter($"level" === 5)
      .select($"name", rule_to_tuple($"rule").as("rules")
      )
      .select(
        $"name",
        $"rules._1".as("start"),
        $"rules._2".as("end")
      )
    // 3. 返回标签规则
    ruleDF
  }

  /**
   * 将属性标签数据中[规则: rule与名称：name ]  转化为Map集合
   *
   * @param tagDF
   * @return
   */
  def convertMap(tagDF: DataFrame): Map[String, String] = {
    import tagDF.sparkSession.implicits._
    tagDF.filter($"level" === 5)
      .select($"rule", $"name")
      .as[(String, String)]
      //转为rdd
      .rdd
      // 转为map集合
      .collectAsMap().toMap
  }

  /**
   * 依据[标签业务字段的值]与[标签规则]匹配，进行打标签（userId, tagName)
   * @param dataframe 标签业务数据
   * @param field 标签业务字段
   * @param tagDF 标签数据
   * @return 标签模型数据
   */
  def ruleMatchTag(dataFrame: DataFrame, field: String, tagDf: DataFrame): Unit = {
    val spark: SparkSession = dataFrame.sparkSession
    import spark.implicits._
    // 1. 获取规则rule与tagName集合
    val attrTagRuleMap: Map[String, String] = convertMap(dataFrame)

    //2. map 集合数据广播出去
    val attrTagRuleMapBroadcast = spark.sparkContext.broadcast(attrTagRuleMap)
    // 3. 自定义UDF函数, 依据Job职业和属性标签规则进行标签化
    val field_to_tag: UserDefinedFunction = udf(
      (field: String) => attrTagRuleMapBroadcast.value(field)
    )

    // 4. 计算标签，依据业务字段值获取标签ID
    val modelDF: DataFrame = dataFrame
      .select(
        $"id".as("userId"), //
        field_to_tag(col(field)).as(field)
      )
    modelDF.printSchema()
    modelDF.show(50, truncate = false)

    // 5. 返回计算标签数据
    modelDF
  }


}
